#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <limits.h>
#include <dirent.h>
#include <sys/time.h>
#include <sys/types.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/socket.h>

#include "data_type.h"
#include "cube.h"
#include "cube_define.h"
#include "cube_record.h"

#include "file_struct.h"
#include "user_define.h"
#include "node_define.h"
#include "data_define.h"

#include "data_client.h"
extern struct timeval time_val;

char Data_Buf[2048];
int max_view_data = 30;

RECORD(DATA_DEFINE,ENV_DATA) env_data;

int data_client_init(void * sub_proc,void * para)
{
	int ret;
    DIR * dir;    
	// add youself's plugin init func here

    // create env_data dir to store download data
    dir = opendir("env_data");
    if(dir==NULL)
        system("mkdir env_data");

	return 0;
}

int data_client_start(void * sub_proc,void * para)
{
	int ret;
	int i;
	void * recv_msg;
	int type;
	int subtype;
	// add yorself's module exec func here

	while(1)
	{
		usleep(time_val.tv_usec);
		ret=ex_module_recvmsg(sub_proc,&recv_msg);
		if(ret<0)
			continue;
		if(recv_msg==NULL)
			continue;
		type=message_get_type(recv_msg);
		subtype=message_get_subtype(recv_msg);
		if(!memdb_find_recordtype(type,subtype))
		{
			printf("message format (%d %d) is not registered!\n",
			message_get_type(recv_msg),message_get_subtype(recv_msg));
			continue;
		}
		if((type==TYPE(DATA_DEFINE))&&(subtype==SUBTYPE(DATA_DEFINE,DOWNLOAD_REQ)))
		{
			ret=proc_data_client_download_deal(sub_proc,recv_msg);
		}
        
	}
	return 0;
}

int proc_data_client_download_deal(void * sub_proc,void * recv_msg)
{
	int ret;
	int i;
    RECORD(DATA_DEFINE,DOWNLOAD_REQ) * download_req;
    RECORD(DATA_DEFINE,ENV_DATA) * env_data;
    MSG_EXPAND * msg_expand;

	ret=message_get_record(recv_msg,&download_req,0);
	if(ret<0)
		return ret;

    if((download_req->data_num <0) ||(download_req->data_num >250))
    {
        print_cubeerr("data_client: invalid data num %d, should be 0-250",download_req->data_num);
        return -EINVAL;
    }
    
    // judge it is req message or return message
    ret = message_get_define_expand(recv_msg,&msg_expand,TYPE_PAIR(DATA_DEFINE,ENV_DATA));
    if(msg_expand == NULL)
    {
        // it is req message, process req message
        ret = proc_data_client_datareq(sub_proc,recv_msg,download_req);
    }
    else
    {
        // it is return message, process return message
        ret = proc_data_client_datareturn(sub_proc,recv_msg,download_req);
    }
    return ret;
}

int proc_data_client_datareq(void * sub_proc, void * recv_msg,
    RECORD(DATA_DEFINE,DOWNLOAD_REQ) * download_req)
{
    // req message process function:
    // fill req record, store it to module context
    // and send req message 
	RECORD(USER_DEFINE,CLIENT_STATE) * client_state;
    
	client_state = proc_share_data_getpointer();
    download_req->user=dup_str(client_state->name,DIGEST_SIZE);
    
   // ex_module_setpointer(sub_proc,download_req);
    void * new_msg = message_create(TYPE_PAIR(DATA_DEFINE,DOWNLOAD_REQ),recv_msg);
    if(new_msg == NULL)
        return -EINVAL;

    message_add_record(new_msg,download_req);
    return ex_module_sendmsg(sub_proc,new_msg);
}


int proc_data_client_datareturn(void * sub_proc, void * recv_msg,
    RECORD(DATA_DEFINE,DOWNLOAD_REQ) * download_req)
{
    int ret;
    RECORD(DATA_DEFINE,ENV_DATA) * env_data;
    RECORD(DATA_DEFINE,DATA_UNIT) data_unit;

    RECORD(DATA_DEFINE,DATA_VIEW) * data_view = NULL;
    MSG_EXPAND * msg_expand;
    char NameBuf[64];

    int data_offset = 0;
    int env_data_offset = 0;
    int unit_num=0;
    const int wait_time = 500000;
    const int wait_interval = 50000;
    void * view_msg =NULL;
    int sleep;
    int fd;

    ret=message_remove_expand(recv_msg,TYPE_PAIR(DATA_DEFINE,ENV_DATA),&msg_expand);
    while(msg_expand !=NULL)
    {
        env_data=msg_expand->expand;
        if(env_data == NULL)
            return -EINVAL;
        // rebuild env_data's data file name
        Memset(NameBuf,0,DIGEST_SIZE*3);
        Strcpy(NameBuf,"env_data/");
        digest_to_uuid(env_data->data_uuid,NameBuf+Strlen(NameBuf));
        
        for(sleep=0;sleep<wait_time/wait_interval;sleep++)
        {
            usleep(wait_interval);
            fd=open(NameBuf,O_RDONLY);
            if(fd<0)
                continue;
            ret=read(fd,Data_Buf,250*sizeof(int));
            close(fd);
            if(ret>0)
                break;
        }
        if(sleep>wait_time/wait_interval)
            return -EINVAL;
    
        env_data_offset = (download_req->start_time-env_data->start_time)*1000/env_data->interval/50;

        void * unit_template = memdb_get_template(TYPE_PAIR(DATA_DEFINE,DATA_UNIT));

        // convert env_data to dataview record
        while((env_data_offset < env_data->data_num)
                && (data_offset <download_req->data_num))
        {
            if(unit_num == 0)
            {
                // alloc a dataview and fill the parameter
                data_view = Talloc0(sizeof(*data_view));
                if((download_req->data_num - data_offset) >= max_view_data)
                {
                    data_view->data_num = max_view_data;
                }
                else
                {
                    data_view->data_num = download_req->data_num - data_offset;
                }
                data_view->start_time = env_data->start_time+env_data_offset *env_data->interval*50/1000;
                data_view->interval = env_data->interval;
                data_view->data_array = (void * )Talloc0(32*data_view->data_num);
                if(data_view->data_array == NULL)
                    return -ENOMEM;
            }
            // get a unit and fill it
            data_unit.serial_no = data_offset;
            data_unit.value     = *(int *)(Data_Buf+sizeof(int)*env_data_offset);
            env_data_offset++;
            data_offset++;

                
            // generate data info string
            struct_2_json(&data_unit,data_view->data_array+unit_num*32,unit_template);    
            unit_num++;
            // data view is full, prepare to send it    
            if(unit_num >= data_view->data_num)
            {
                view_msg = message_create(TYPE_PAIR(DATA_DEFINE,DATA_VIEW),recv_msg);
                if(view_msg == NULL)
                    return -EINVAL;
                message_add_record(view_msg,data_view);
                ex_module_sendmsg(sub_proc,view_msg);
                unit_num=0;
                view_msg=NULL;
            }
        }    
        ret=message_remove_expand(recv_msg,TYPE_PAIR(DATA_DEFINE,ENV_DATA),&msg_expand);
    }   

    if(unit_num >0)
    {
        view_msg = message_create(TYPE_PAIR(DATA_DEFINE,DATA_VIEW),recv_msg);
        if(view_msg == NULL)
            return -EINVAL;
        message_add_record(view_msg,data_view);
        ex_module_sendmsg(sub_proc,view_msg);
    }

    return ret;
}
